package com.heima.kafka.simple;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;
import java.util.concurrent.ExecutionException;

public class ConsumerTest {

    public static void main(String[] args) {
        //1、配置kafka
        Properties properties = new Properties();
        //指定kafka服务器地址
        properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.200.130:9092");
        //指定反序列化器
        properties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringDeserializer");
        properties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringDeserializer");
        //指定消费者所属的组
        properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG,"group002");

        //指定偏移量的提交方式
        //指定手动提交偏移量
        properties.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,"false");



        //2、创建消费者
        KafkaConsumer<String,String> consumer = new KafkaConsumer<>(properties);

        //3、订阅主题
        consumer.subscribe(Collections.singleton("topicB"));

        //4、从kafka拉取消息
        while (true) {

                ConsumerRecords<String, String> messages = consumer.poll(Duration.ofMillis(1000));

            for (ConsumerRecord<String, String> message : messages) {
                System.out.println(message.key()+"=====>"+message.value());
            }
        }
    }
}
